﻿/******************************************************************************
 * Copyright (C) 2015  671643387@qq.com
 *
 * This program is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License as published by the
 * Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
 * more details.
 *
 * You should have received a copy of the GNU General Public License along
 * with this program; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
 *
 *****************************************************************************/

#include "Shared/Base/ThreadPool.h"
#include <memory>
#include "Shared/Base/Runnable.h"
#include "Shared/Base/Event.h"
#include "Shared/Base/LockGuard.h"

#pragma warning(disable:4355)


namespace Shared {

    class PooledThread: private Runnable
    {
    public:
        PooledThread(ThreadPool &pool)
            : _pool(pool)
            , _thread(this)
            , _target(NULL)
            , _targetReady()
            , _started()
            , _targetLock()
            , _busy(false)
        {

        }

        virtual ~PooledThread(void)
        {
            stop();
            join();
        }

        bool start(void)
        {
            if (_thread.start()) {
                _thread.setPriority(_pool._priority);
                _started.wait();
                return true;
            }
            return false;
        }

        void stop(void)
        {
            SHARED_LOCK_GURAD_ASSERT(_targetLock);
            _target = NULL;
            _targetReady.set();
        }

        void join(void)
        {
            _thread.join();
        }

        bool start(Runnable *runnable)
        {
            SHARED_ASSERT(runnable);
            SHARED_LOCK_GURAD_ASSERT_RETURN_VAR(_targetLock, false);
			_busy = true;
            _target = runnable;
            _targetReady.set();
            return true;
        }

        bool start(Runnable *runnable, int priority)
        {
            SHARED_ASSERT(runnable);
            SHARED_LOCK_GURAD_ASSERT_RETURN_VAR(_targetLock, false);
			_busy = true;
            _target = runnable;
            _thread.setPriority(priority);
            _targetReady.set();
            return true;
        }

        virtual void run(void)
        {
            _started.set();
            for (;;)
            {
				_busy = false;
                _targetReady.wait();
                SHARED_LOCK_GURAD_ASSERT(_targetLock);
                if (_target)
                {
                    _target->run();
                    continue;
                }
                break;
            }
        }

        bool busy(void) const
        {
            return _busy;
        }

    private:
        ThreadPool &_pool;
        Thread _thread;
        Runnable *_target;
        Event _targetReady;
        Event _started;
        Mutex _targetLock;
        volatile bool _busy;
        DISALLOW_COPY_AND_ASSIGN(PooledThread);
    };


    ThreadPool::ThreadPool(int initThread, int maxThread, int priority /* = Thread::PRIO_NORMAL */)
        : _initThreads(initThread)
        , _maxThreads(maxThread)
        , _priority(priority)
        , _threads()
        , _threadsLock()
        , _inited(false)
		, _joined(false)
		, _stopping(false)
    {

    }

    ThreadPool::~ThreadPool(void)
    {
        uninitialize();
    }

    bool ThreadPool::start(void)
    {
        SHARED_ASSERT(!_inited);
        _inited = true;

        for (int newThread = 0; 
            newThread < _initThreads; ++newThread) {
                if (!createThread()) {
                    break;
                }
        }

        return _threads.size() == _initThreads;
    }

    void ThreadPool::stop(void)
    {
		SHARED_ASSERT(!_stopping);
		_stopping = true;

		if (_joined)
		{
			PooledThreadQueue::iterator iter = _threads.begin();
			for (; iter != _threads.end(); ++iter) {
				(*iter)->stop();
			}
		} else {
			SHARED_LOCK_GURAD_ASSERT(_threadsLock);
			PooledThreadQueue::iterator iter = _threads.begin();
			for (; iter != _threads.end(); ++iter) {
				(*iter)->stop();
				(*iter)->join();
			}
		}
    }

    void ThreadPool::join(void)
    {
		SHARED_ASSERT(!_joined);
		_joined = true;
        SHARED_LOCK_GURAD_ASSERT(_threadsLock);
        PooledThreadQueue::iterator iter = _threads.begin();
        for (; iter != _threads.end(); ++iter) {
            (*iter)->join();
        }
    }

    bool ThreadPool::start(Runnable *runnable)
    {
        SHARED_ASSERT(runnable && !_stopping);
        PooledThread *workerThread = getThread();
        if (workerThread) {
            return workerThread->start(runnable);
        }
        return false;
    }

    bool ThreadPool::start(Runnable *runnable, int priority)
    {
        SHARED_ASSERT(runnable && !_stopping);
        PooledThread *workerThread = getThread();
        if (workerThread) {
            return workerThread->start(runnable, priority);
        }
        return false;
    }

	bool ThreadPool::start(Runnable **runables, int nRunnables)
	{
		SHARED_ASSERT(runables && !_stopping);
		for (int index = 0; index < nRunnables; ++index)
		{
			PooledThread *workerThread = getThread();
			if (!(workerThread 
				&& workerThread->start(runables[index]))) {
				return false;
			}
		}
		return true;
	}

	void ThreadPool::uninitialize(void)
	{
		while (!_threads.empty()) {
			delete _threads.front();
			_threads.pop_front();
		}
	}

    PooledThread *ThreadPool::createThread(void)
    {
        PooledThread *newThread = new PooledThread(*this);
        std::auto_ptr<PooledThread> newPooledThread(newThread);
        if (newThread && newThread->start()) {
            _threads.push_back(newPooledThread.release());
            return newThread;
        }
        return NULL;
    }

    PooledThread *ThreadPool::getThread(void)
    {
        SHARED_LOCK_GURAD_ASSERT_RETURN_VAR(_threadsLock, NULL);
        PooledThread *workerThread = NULL;
        PooledThreadQueue::iterator iter = _threads.begin();
        for (; iter != _threads.end();) {
            if (!(*iter)->busy()) {
                if ((_threads.size() > (std::size_t)_maxThreads) 
                    && (workerThread != NULL)) {
                        (*iter)->stop();
                        (*iter)->join();
                        iter = _threads.erase(iter);
                        continue;
                } else if (_threads.size() > (std::size_t)_maxThreads) {
                    workerThread = *(iter);
                } else {
                    return *(iter);
                }
            }
            ++iter;
        }
        return createThread();
    }

    bool ThreadPool::addCapacity(int threads)
    {
		SHARED_ASSERT(!_stopping);
        if (!_inited) {
            _initThreads += threads;
            _maxThreads += threads;
        } else {
			SHARED_LOCK_GURAD_ASSERT_RETURN_VAR(_threadsLock, false);
            _maxThreads += threads;
            for (int nThread = 0; 
                nThread < threads; ++nThread)
            {
                if (!createThread()) {
                    return false;
                }
            }
        }
        return true;
    }
}